package io.kiki.sba.registry.server.clientInterface.multi.cluster;

import com.google.common.collect.Sets;
import io.kiki.sba.registry.common.model.metaserver.MultiClusterSyncInfo;
import io.kiki.sba.registry.common.model.multi.cluster.DataCenter;
import io.kiki.sba.registry.common.model.multi.cluster.RemoteSlotsStatus;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.store.api.meta.MultiClusterSyncRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


public class DataCenterCacheImpl implements DataCenterCache {

    private static final Logger logger = LoggerFactory.getLogger(DataCenterCacheImpl.class);

    @Autowired
    private Config config;

    @Autowired
    private MultiClusterSyncRepository multiClusterSyncRepository;

    private Map<String, DataCenter> dataCenterToDataCenterMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // init local dataCenter zones
        dataCenterToDataCenterMap.put(config.getDataCenter(), new DataCenter(config.getDataCenter(), config.getLocalDataCenterZones()));
    }


    @Override
    public Set<String> getZoneSet(String dataCenter) {
        DataCenter dataCenter1 = dataCenterToDataCenterMap.get(dataCenter);

        if (dataCenter1 == null) {
            return null;
        }
        return dataCenter1.getZoneSet();
    }

    @Override
    public Map<String, Set<String>> getZoneSet(Set<String> dataCenterSet) {
        if (CollectionUtils.isEmpty(dataCenterSet)) {
            return Collections.EMPTY_MAP;
        }

        Map<String, Set<String>> map = new HashMap<>(dataCenterSet.size());
        for (String dataCenter : dataCenterSet) {
            DataCenter dataCenter1 = dataCenterToDataCenterMap.get(dataCenter);
            if (dataCenter1 == null || CollectionUtils.isEmpty(dataCenter1.getZoneSet())) {
                logger.error("[DataCenterMetadataCache]find dataCenter: {} zones error.", dataCenter);
                continue;
            }
            map.put(dataCenter, dataCenter1.getZoneSet());
        }
        return map;
    }

    @Override
    public boolean saveDataCenterList(Map<String, RemoteSlotsStatus> dataCenterToRemoteSlotsStatusMap) {

        Set<String> set = new HashSet<>(dataCenterToRemoteSlotsStatusMap.keySet());
        set.add(config.getDataCenter());
        Set<String> tobeRemove = Sets.difference(dataCenterToDataCenterMap.keySet(), set);

        boolean success = true;
        for (Entry<String, RemoteSlotsStatus> entry : Optional.ofNullable(dataCenterToRemoteSlotsStatusMap).orElse(new HashMap<>()).entrySet()) {

            RemoteSlotsStatus remoteSlotsStatus1 = entry.getValue();
            if (StringUtils.isEmpty(entry.getKey()) || remoteSlotsStatus1.getDataCenter() == null || CollectionUtils.isEmpty(remoteSlotsStatus1.getDataCenter().getZoneSet())) {
                logger.error("[DataCenterMetadataCache]invalidate dataCenter: {} or metadata: {}", entry.getKey(), remoteSlotsStatus1);
                success = false;
                continue;
            }

            DataCenter exist = dataCenterToDataCenterMap.get(entry.getKey());
            if (!remoteSlotsStatus1.equals(exist)) {
                dataCenterToDataCenterMap.put(entry.getKey(), remoteSlotsStatus1.getDataCenter());
                logger.info("[DataCenterMetadataCache]update from:{} to:{}", exist, remoteSlotsStatus1);
            }
        }

        processRemove(tobeRemove);
        return success;
    }

    private void processRemove(Set<String> tobeRemoveDataCenterSet) {
        if (CollectionUtils.isEmpty(tobeRemoveDataCenterSet)) {
            return;
        }
        Set<MultiClusterSyncInfo> set = multiClusterSyncRepository.queryLocalMultiClusterSyncInfoSet();
        Set<String> syncing = set.stream().map(MultiClusterSyncInfo::getRemoteDataCenter).collect(Collectors.toSet());
        for (String dataCenter : tobeRemoveDataCenterSet) {
            if (syncing.contains(dataCenter)) {
                logger.error("dataCenter:{} remove metadata is forbidden.", dataCenter);
                continue;
            }
            dataCenterToDataCenterMap.remove(dataCenter);
            logger.info("remove dataCenter:{} metadata success.", dataCenter);
        }
    }

    @Override
    public Set<String> getSyncDataCenters() {
        return dataCenterToDataCenterMap.keySet();
    }

    @Override
    public DataCenter of(String dataCenter) {
        DataCenter dataCenter1 = dataCenterToDataCenterMap.get(dataCenter);
        return dataCenter1;
    }


}
